package cn.doitedu.dao;

import cn.doitedu.pojo.LogBean;
import cn.doitedu.utils.ConnectionUtils;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Table;

import java.util.Map;
import java.util.Set;

/**
 * @create: 2021-11-13 19:18
 * @author: 静如海的男人
 * @program: eagle-markting
 *      匹配用户画像
 **/

public class ProfileHbaseDaoImpl implements ProfileDao {
    private Connection hbase;
    //初始化方法
    @Override
    public void init(ParameterTool parameterTool) throws Exception {
        hbase = ConnectionUtils.createHbase(parameterTool);
    }
    //匹配用户画像{tag1 = v1, tag10 = v1}
    @Override
    public boolean isMatchProfile(LogBean bean, Map<String, String> profilemap) throws Exception {

        if (profilemap != null && profilemap.size() > 0) {
            Table table = hbase.getTable(TableName.valueOf("eagle_profile"));
            Get get = new Get(bean.getDeviceId().getBytes());
            //定义好的规则key
            Set<String> keySet = profilemap.keySet();
            for (String key : keySet) {
                get.addColumn("f".getBytes(), key.getBytes());
            }
            //加入定义好的规则请求Hbase
            Result result = table.get(get);
            //查询结果如果为空直接false
            if (result != null) {
                for (String key : keySet) {
                    //请求Hbase返回的数据有可能为null 如果为null直接false
                    byte[] value = result.getValue("f".getBytes(), key.getBytes());
                    if (value != null) {
                        //定义好的规则
                        String pv = profilemap.get(key);
                        if (!pv.equals(new String(value))) {
                            return false;
                        }
                    } else {
                        return false;
                    }
                }
            } else {
                return false;
            }
        }

        return true;
    }

    @Override
    public void close() throws Exception {
        hbase.close();
        ConnectionUtils.close();
    }
}
